package com.sleep.ws.client.core;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
import io.netty.handler.codec.http.websocketx.WebSocketVersion;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.internal.StringUtil;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import javax.net.ssl.TrustManager;
import javax.net.ssl.X509TrustManager;
import java.net.URI;
import java.security.cert.X509Certificate;
import java.util.concurrent.TimeUnit;

/**
 * websocket客户端，接收消息
 * @author Sleep-6000
 * @date 2024/3/8
 */
@Slf4j
public class WsClient {

    private final URI uri;

    private Bootstrap b = new Bootstrap();

    private Channel ch;

    private WsClientHandler handler;

    private EventLoopGroup group = new NioEventLoopGroup(5);

    private String productToken = "";

    @Setter
    private int reconnectTime = 10;

    public WsClient(String url) {
        this.uri = URI.create(url);
        String protocol = uri.getScheme();
        b.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new IdleStateHandler(10, 10, 0, TimeUnit.SECONDS));
                        //TODO client提供set证书功能，然后在这里进行certificates校验
                        if ("wss".equals(protocol)) {
                            TrustManager[] trustAllCerts = new TrustManager[]{
                                    new X509TrustManager() {
                                        @Override
                                        public void checkClientTrusted(X509Certificate[] x509Certificates, String s) {
                                        }

                                        @Override
                                        public void checkServerTrusted(X509Certificate[] x509Certificates, String s) {
                                        }

                                        @Override
                                        public X509Certificate[] getAcceptedIssuers() {
                                            return new X509Certificate[]{};
                                        }
                                    }
                            };
                            //初始化ssl相关配置
                            SSLContext sslContext = SSLContext.getInstance("TLS");
                            sslContext.init(null, trustAllCerts, null);
                            SSLEngine sslEngine = sslContext.createSSLEngine();
                            sslEngine.setUseClientMode(true);
                            pipeline.addLast(new SslHandler(sslEngine));
                        }
                        WsClientHandler currentHandler = new WsClientHandler(
                                WebSocketClientHandshakerFactory.newHandshaker(
                                        uri, WebSocketVersion.V13, null, true, HttpHeaders.EMPTY_HEADERS, 10240),
                                WsClient.this);

                        handler = currentHandler;
                        pipeline.addLast(new HttpClientCodec());
                        pipeline.addLast(new HttpObjectAggregator(65536));
                        pipeline.addLast(currentHandler);
                    }
                });
        open();
    }

    public WsClient(String uri, String productToken) {
        this(uri);
        this.productToken = productToken;
    }


    public void open() {
        try {
            ChannelFuture future = b.connect(uri.getHost(), uri.getPort());
            future.addListener((ChannelFutureListener) channelFuture -> {
                if (!channelFuture.isSuccess()) {
                    channelFuture.channel().eventLoop().schedule(() -> {
                        log.error("websocket连接服务端失败，将重新尝试连接");
                        open();
                    }, reconnectTime, TimeUnit.SECONDS);
                } else {
                    log.info("服务端链接成功...");
                    handler.reconnectStatus = true;
                    ch = channelFuture.channel();
                }
            });
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

    public void close() {
        try {
            log.info("channel关闭");
            handler.reconnectStatus = false;
            ch.writeAndFlush(new CloseWebSocketFrame());
            ch.closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("channel关闭失败:{}", e.getMessage());
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 根据topic发送消息
     *
     * @param topic 生产者
     * @param msg   消息
     */
    public void eval(String topic, final Object msg) {
        if (StringUtil.isNullOrEmpty(topic)) {
            log.error("Topic为空，无法推送数据：topic: {}, msg: {}", topic, msg);
            return;
        }
        if (ch != null && ch.isActive()) {
            try {
                ch.writeAndFlush(new TextWebSocketFrame(PushMessageTemplate.generateTemplate(topic, msg, productToken)));
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        } else {
            log.error("Websocket通道失效，无法推送数据：topic: {}, msg: {}", topic, msg);
        }
    }

}
